package net.bwie.realtime.vehicle2.function;

import net.bwie.realtime.vehicle2.bean.VehicleData;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author: FuHe
 * @Date: 2025/6/8
 */
public class VehicleRealtimeStatusFunction extends KeyedProcessFunction<String, VehicleData, String> {
    // 声明状态变量  用于存储上一次发送的时间
    ValueState<Long> lastTimestamp;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态变量
        lastTimestamp = getRuntimeContext().getState(
                new ValueStateDescriptor<>("last-timestamp", Long.class)
        );
    }

    @Override
    public void processElement(VehicleData value, Context ctx, Collector<String> out) throws Exception {
        // 获取状态变量值
        Long lastEmitTime = lastTimestamp.value();
        // 当前数据的时间
        long currentTime = ctx.timerService().currentProcessingTime();
        // 判断是否满足输出条件
        if (lastEmitTime == null || currentTime - lastEmitTime >= 3000) {
            lastTimestamp.update(currentTime);
            out.collect(value.getVin() + "," +
                    value.getSpeed() + "," +
                    value.getBatteryLevel() + "," +
                    value.getBatteryTemp() + "," +
                    value.getMotorTemp() + "," +
                    value.getChargingStatus() + "," +
                    value.getEnergyConsumption());
        }
    }
}
